package com.za.kafka.naive;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

/**
 * Kafka Consumer
 * <p>
 * 偏移量
 * Kafka为分区中的每个记录保留一个数值偏移量。这个偏移量充当该分区中记录的唯一标识符，并且表示该分区中的消费者的位置。
 * 也就是说，拥有5号位置的消费者使用偏移值为0到4的记录，并将使用偏移量5记录下一个记录。实际上，与使用者的使用者有关的位置有两个概念。
 * 消费者的位置将提供下一个记录的偏移量。它将比消费者在该分区中看到的最高偏移量大一个。它在消费者每次接收数据调用poll(long)和接收消息时自动地前进。
 * 已提交的位置是安全保存的最后一个偏移量。如果进程失败并重新启动，这将恢复到它将恢复的偏移量。消费者可以定期自动提交补偿;
 * 或者，它可以选择通过调用commitSync来手动控制这个提交的位置，它会阻塞，直到在提交过程中成功提交了补偿或致命错误，
 * 或者提交了非阻塞的commitAsync，并将触发OffsetCommitCallback，要么成功提交，要么失败。
 * <p>
 * 消费者组和订阅主题
 * Kafka使用了消费者组（Consumer Groups）的概念来允许一个过程池来划分消费和处理记录的工作。这些进程可以在同一台机器上运行，
 * 或者，更有可能的是，它们可以分布在许多机器上，从而为处理提供额外的可伸缩性和容错。
 * 每个Kafka消费者可以配置一个消费者团体,它属于,并且可以动态设置主题希望通过订阅订阅列表(列表,Consumer Rebalance Listener),
 * 或通过订阅订阅所有主题匹配特定的模式(模式,ConsumerRebalanceListener)。Kafka将在每个用户组中的一个流程中传递订阅主题中的每个消息。
 * 这是通过在每个组的消费者过程中平衡主题中的分区来实现的。因此，如果有一个带有四个分区的主题，以及一个有两个进程的消费者组，
 * 每个进程将从两个分区中消耗。这个组成员是动态维护的:如果进程失败，分配给它的分区将被重新分配给同一组中的其他进程，
 * 如果一个新进程加入这个组，那么分区将从现有的消费者转移到这个新的进程。
 * 因此，如果两个进程订阅一个主题同时指定不同的组，则它们将在该主题中获得所有的记录;如果他们都指定相同的组，他们将会得到大约一半的记录。
 * 从概念上来说，您可以将消费者组看作是由多个进程组成的单个逻辑订阅者。作为一个多用户系统，Kafka自然支持在没有重复数据的情况下为给定的主题
 * 提供任意数量的用户组(额外的消费者实际上相当便宜)。这是对消息传递系统中常见的功能的轻微概括。
 * 为了获得与传统消息传递系统中的队列相似的语义，所有流程都将是单个消费者组的一部分，因此记录交付将在组中与队列相同。
 * 与传统的消息传递系统不同，您可以拥有多个这样的组。要获得类似于传统消息传递系统中的pub-sub的语义，每个流程都有自己的消费者组，
 * 因此每个进程将订阅发布到该主题的所有记录。
 * 此外,自动组分配时,消费者可以通过Consumer Rebalance Listener通知,这允许他们完成清理等必要的应用程序级的逻辑状态,
 * 手动抵消提交(注意,抵消总是承诺对于一个给定的消费者团体),等等的详细信息,
 * 请参阅存储抵消外部卡夫卡对消费者也可以手动指定分区分配给它分配(列表),禁用这个动态分区分配。
 *
 * @author tongyufu
 */
public class MyConsumer {

    String groupId = "group-1";
    String servers = "192.168.31.134:9092";
    String topic = "test";

    public static void main(String[] args) {
        MyConsumer consumer = new MyConsumer();
        //consumer.automaticOffsetCommitting();
        consumer.manualOffsetControl();
    }

    /**
     * 自动确认Offset
     */
    void automaticOffsetCommitting() {
        Properties props = new Properties();
        props.put("bootstrap.servers", servers);
        /*
         * 用来唯一标识consumer进程所在组的字符串，如果设置同样的group id，表示这些processes都是属于同一个consumer
         * group，这些Consumer将分享消息。 不同group的consumer，每个group中的consumer互不干扰，都可以获得完整的消息。
         * 同一个group中的Consumer数量不要大于Topic的partitions值，超出部分将无法消费到任何消息（多线程同理）
         */
        props.put("group.id", groupId);
        // 如果为true，consumer所fetch的消息的offset将会自动的同步到zookeeper。这项提交的offset将在进程挂掉时，由新的consumer使用
        props.put("enable.auto.commit", "true");
        // consumer向zooKeeper提交offset的频率，单位是毫秒
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        System.out.println("===消费组：" + groupId);

        KafkaConsumer<String, String> consumer = null;
        try {
            consumer = new KafkaConsumer<>(props);
            // 消费者订阅的topic, 可同时订阅多个
            consumer.subscribe(Arrays.asList(topic));
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records)
                    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(),
                            record.value());
            }
        } finally {
            if (consumer != null) {
                consumer.close();
            }
        }
    }

    /**
     * 手工控制Offset
     */
    void manualOffsetControl() {
        KafkaConsumer<String, String> consumer = null;
        Properties props = new Properties();
        props.put("bootstrap.servers", servers);
        props.put("group.id", groupId);
        // 关闭自动确认
        props.put("enable.auto.commit", "false");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        final int minBatchSize = 2;
        List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
        try {
            consumer = new KafkaConsumer<>(props);
            TopicPartition topicPartition = new TopicPartition(topic, 0);
            // 指定消费topic的那个分区
            consumer.assign(Arrays.asList(topicPartition));
            // 指定offset
            consumer.seek(topicPartition, 5);
            //consumer.subscribe(Arrays.asList(topic));
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records) {
                    buffer.add(record);
                    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(),
                            record.value());
                }
                // 数据达到批量要求，就写入DB，同步确认offset
                if (buffer.size() >= minBatchSize) {
                    System.out.println("数据达到批量要求，同步确认offset");
                    // insertIntoDb(buffer);
                    consumer.commitSync();
                    buffer.clear();
                }
            }
        } finally {
            if (consumer != null) {
                consumer.close();
            }
        }
    }

}
